package my.thesis.bolts.aggregatorsReliable;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import my.thesis.analytics.metrics.ThroughputMetric;
import my.thesis.utilities.Tags;

import org.json.JSONException;

import backtype.storm.Config;
import backtype.storm.Constants;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class HashtagCounterReliable extends BaseRichBolt
{
private static final long serialVersionUID = 1L;
	
	private ConcurrentHashMap<Long, HashMap<String, HashMap<String, Integer>>> countsPerTimeIntervalPerLanguage;
	private ConcurrentHashMap<Long,ArrayList<Tuple>> tuplesToAck;
	private OutputCollector collector;
	private int secToEmit;
	private Long emitUpToTimeInterval;
	
	private Boolean receivedSinceLastEmit;
	transient ThroughputMetric _throughputMetric;
	private static Boolean metricsOn=false;
	
	public HashtagCounterReliable(int emSec)
	{
		secToEmit=emSec;
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
	    declarer.declare(new Fields("timeInterval", "countsForInterval"));
	}
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector)
	{
		countsPerTimeIntervalPerLanguage = new ConcurrentHashMap<Long, HashMap<String, HashMap<String, Integer>>>();
		this.collector=collector;
		tuplesToAck = new ConcurrentHashMap<Long,ArrayList<Tuple>>();
		
		this.receivedSinceLastEmit=false;
		
		if(metricsOn)
		{
			_throughputMetric=new ThroughputMetric();
			context.registerMetric("topHashtagCount", _throughputMetric, 20);
			_throughputMetric.setStartTime();
		}
	}
	
	@Override
	public Map<String, Object> getComponentConfiguration() 
	{
	    Config conf = new Config();
	    conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, secToEmit);
	    return conf;
	}
	
	public void execute(Tuple tuple) 
	{		
		if (isTickTuple(tuple)) 
		{
			if(this.receivedSinceLastEmit)
			{				
				emitHashtags();
			}
			
			this.receivedSinceLastEmit=false;
	    } 
		else 
	    {
			this.receivedSinceLastEmit=true;
					
			if(metricsOn)
			{
				_throughputMetric.incrTuples();
			}
			
			String hash = tuple.getStringByField("hash");
			
			long ti = tuple.getLongByField("timeInterval");
			emitUpToTimeInterval=ti;
			
			String lang = tuple.getStringByField("lang");

			HashMap<String, Integer> countForIntervalForLang = getMapForIntervalForLang(ti,lang);
			countForIntervalForLang.put(hash, countForIntervalAndLangAndHash(ti,lang,hash) + 1);	
			
			//RELIABLE
			ArrayList<Tuple> tuplesToAckForTi = getListForInterval(ti);
			tuplesToAckForTi.add(tuple);
		 }
	}
				  
    private HashMap<String, Integer> getMapForIntervalForLang(Long timeInterval,String lang) 
    {
    	try
    	{
    	  HashMap<String, HashMap<String, Integer>> countForInterval = countsPerTimeIntervalPerLanguage.get(timeInterval);
		    
		  if (countForInterval == null) 
		  {
		    	countForInterval=new HashMap<String, HashMap<String, Integer>>();
		    	HashMap<String,Integer> countForIntervalForLang=new HashMap<String, Integer>();
		    	countForInterval.put(lang, countForIntervalForLang);
		    	countsPerTimeIntervalPerLanguage.put(timeInterval, countForInterval);
		    	return countForInterval.get(lang);
		  }
		  else 
		  { 
			  if(countForInterval.containsKey(lang))
				  return countForInterval.get(lang);
			  else
			  {
				  HashMap<String,Integer> countForIntervalForLang=new HashMap<String, Integer>();
				  countForInterval.put(lang, countForIntervalForLang);
				  return countForInterval.get(lang);
			  }
		  }
    	}
    	catch(Exception ex)
    	{
    		System.out.println(ex.toString());
    		return null;
    	}
	}
    
	private Integer countForIntervalAndLangAndHash(Long timeInterval,String lang,String hash) 
	{
		 Integer count = countsPerTimeIntervalPerLanguage.get(timeInterval).get(lang).get(hash);
		 return count == null ? 0 : count;	
	}
	    
    private boolean isTickTuple(Tuple tuple)
	{
		 String sourceComponent = tuple.getSourceComponent();
		 String sourceStreamId = tuple.getSourceStreamId();
		 return sourceComponent.equals(Constants.SYSTEM_COMPONENT_ID) 
				 && sourceStreamId.equals(Constants.SYSTEM_TICK_STREAM_ID);
	}
    
    private void emitHashtags() 
    {
      //TEST
      System.out.println("EMIT UP TO " + emitUpToTimeInterval);
      
      Set<Long> timeIntervalsAvailable = countsPerTimeIntervalPerLanguage.keySet();
      
      for (Long timeInterval : timeIntervalsAvailable) 
      {
        if (timeInterval < emitUpToTimeInterval) 
        {
        	//TEST
        	System.out.println("Emmiting Hashtags...");
        	
        	HashMap<String, HashMap<String, Integer>> hashtagsForTimeInterval = countsPerTimeIntervalPerLanguage.remove(timeInterval);
        	
        	//************************ TOP-10 **********************************************
        	
        	HashMap<String, HashMap<String, Integer>> hashtagsForTimeIntervalTop10 = new HashMap<String, HashMap<String, Integer>>();
        	for(Entry<String, HashMap<String, Integer>> entry : hashtagsForTimeInterval.entrySet())	
			{
        		 HashMap<String,Integer> hashtagsForLang=entry.getValue();
        		 HashMap<String,Integer> top10Hashtags = getTop10Hashtags(hashtagsForLang);
        		 hashtagsForTimeIntervalTop10.put(entry.getKey(), top10Hashtags);
			}
        	
        		//RELIABLE
        		ArrayList<Tuple> tuplesToAckForTi = tuplesToAck.remove(timeInterval);
        	
        		//multiple anchor
        		collector.emit(tuplesToAckForTi,new Values(timeInterval, hashtagsForTimeIntervalTop10));
        	
	        	//multiple ack
	        	for(Tuple t:tuplesToAckForTi)
	        	{
	        		collector.ack(t);
	        	}
	        	
        	//************************ TOP-10 **********************************************
        }
      }
    }
    	
	private HashMap<String,Integer> getTop10Hashtags(HashMap<String,Integer> tokens)
	{			
			   HashMap<String,Integer> top10Hashtags = new HashMap<String,Integer>();
				
			   Set<String> keys = tokens.keySet();
			   ArrayList<Tags> tags = new ArrayList<Tags>();
			   
			   for(String key:keys)
			   {
			       Tags tag = new Tags();
			       tag.setKey(key);
			       tag.setValue(tokens.get(key));
			       tags.add(tag);
			   }
			   
			   Collections.sort(tags, Collections.reverseOrder());
			   
			   int numwords = 10;
			   if(tags.size()<numwords)
			   {
			       numwords = tags.size();
			   }        
			   for(int i=0;i<numwords;i++)
			   {
			       Tags tag = tags.get(i);
			       try
			       {
			    	   top10Hashtags.put(tag.getKey(), tag.getValue());
			 
			       }
			       catch(JSONException ex)
			       {
			           ex.printStackTrace();
			       }
			   }
			   
			   return top10Hashtags;
	}
	
	private ArrayList<Tuple> getListForInterval(Long ti)
	{
		ArrayList<Tuple> listForInterval=tuplesToAck.get(ti);
		
		if(listForInterval==null)
		{
			ArrayList<Tuple> newList = new ArrayList<Tuple>();
			tuplesToAck.put(ti, newList);
			return tuplesToAck.get(ti);
		}
		else
		{
			return listForInterval;
		}		
	}
}
